package com.shujia.mllib

import org.apache.spark.ml.classification.{LogisticRegression, LogisticRegressionModel}
import org.apache.spark.sql.functions.{count, sum, when}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object DEmo5TrainModel {

  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession.builder()
      .master("local[8]")
      .appName("person")
      .config("spark.sql.shuffle.partitions", 2)
      .getOrCreate()


    /**
      * 1、读取数据
      */


    val data: DataFrame = spark
      .read
      .format("libsvm")
      .load("spark/data/images")

    /**
      * 2、切分训练集和测试集
      *
      */

    val split: Array[Dataset[Row]] = data.randomSplit(Array(0.7, 0.3))

    val train: Dataset[Row] = split(0)
    val test: Dataset[Row] = split(1)

    //构建算法执行参数
    val logisticRegression: LogisticRegression = new LogisticRegression()
      .setMaxIter(10) //最大迭代次数
      .setFitIntercept(true) //是否有截距


    /**
      * 训练模型
      *
      * 通过spark 进行分布式迭代计算
      *
      */

    val model: LogisticRegressionModel = logisticRegression.fit(train)


    /**
      * 4、模型评估
      * 使用模型预测测试集的数据，判断和原始标记是否一直，计算准确率
      *
      */
    val frame: DataFrame = model.transform(test)

    /**
      *
      * 计算准确率
      */

    import spark.implicits._
    import org.apache.spark.sql.functions._

    val result: DataFrame = frame.select(sum(when($"label" === $"prediction", 1).otherwise(0)) / count($"label"))

    result.show()

    /**
      * 保存模型
      *
      */

    model.save("spark/data/image_model")


  }

}
